package com.conversationboard.fulltext.index;

import java.io.File;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;

import org.apache.commons.io.FileUtils;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexWriter;
import org.joda.time.DateTime;
import org.joda.time.MutableDateTime;

import com.conversationboard.config.Configuration;
import com.conversationboard.config.ConnectionStrategy;
import com.conversationboard.database.BatchQueries;
import com.conversationboard.database.Database;
import com.conversationboard.fulltext.DocumentType;
import com.conversationboard.fulltext.data.MessageData;
import com.conversationboard.fulltext.data.ThreadData;
import com.conversationboard.logger.LogMessageType;
import com.conversationboard.logger.Logger;
import com.conversationboard.model.User;

public class IndexNewDocuments {

	private boolean indexAllThreads = false;


	public boolean isIndexAllThreads() {
		return indexAllThreads;
	}


	public void setIndexAllThreads(boolean indexAllThreads) {
		this.indexAllThreads = indexAllThreads;
	}


	public void indexNewDocuments() throws SQLException, IOException {

		this.indexNewThreads();
		this.indexNewMessages();
	}


	/**
	 * Ensure that the query:
	 *  (1) Gets the current time; assign to currentTime
	 *  (2) Writes that as the time when the last indexing took place
	 *  (3) Queries the database to only retrieve between the last indexing and what currentTime is set to
	 */

	private MutableDateTime getNewThreads(List<ThreadData> threads) throws SQLException {

		Connection connection = null;

		try {

			MutableDateTime maxDateTime = new MutableDateTime(1, 1, 1, 1, 1, 1, 1);

			connection = Database.getConnection();
			
			String query = BatchQueries.GET_THREADS_NEEDING_INDEXED;

			if (this.isIndexAllThreads()) {
				query = BatchQueries.GET_ALL_THREADS;
			}
			 
			PreparedStatement statement = connection.prepareStatement(query);

			statement.execute();

			ResultSet resultSet = statement.getResultSet();

			while (resultSet.next()) {

				int boardId = resultSet.getInt("boardid");
				int threadId = resultSet.getInt("threadid");
				String title = resultSet.getString("title");
				String displayName = resultSet.getString("displayname");
				String loginId = resultSet.getString("loginid");
				Timestamp timestamp = resultSet.getTimestamp("timestamp");
				boolean deleted = resultSet.getBoolean("deleted");

				DateTime time = new DateTime(timestamp);

				/* Ensure we keep a record of the most recent new thread */

				if (time.isAfter(maxDateTime)) {
					maxDateTime = time.toMutableDateTime();
				}

				ThreadData thread = new ThreadData(boardId, threadId, title, displayName, loginId, timestamp, deleted);
				threads.add(thread);
			}

			resultSet.close();
			statement.close();

			return maxDateTime;

		} finally {
			Database.close(connection);
		}
	}


	/**
	 * Add all the threads into the Lucene index.
	 * 
	 * @param threads
	 * @throws IOException
	 */

	private void indexThreads(List<ThreadData> threads) throws CorruptIndexException, IOException {

		if (threads.size() == 0) {
			return;
		}

		File indexDirectory = new File(Configuration.getInstance().getThreadIndexPath());
		IndexWriter indexWriter = new IndexWriter(indexDirectory, new StandardAnalyzer());

		try {
			Field id = new Field("id", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field boardId = new Field("boardId", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field title = new Field("text", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field name = new Field("displayName", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field login = new Field("loginId", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field date = new Field("date", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field time = new Field("time", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field deleted = new Field("deleted", "", Field.Store.YES, Field.Index.UN_TOKENIZED);

			SimpleDateFormat formatter = new SimpleDateFormat("yyMMdd");
			SimpleDateFormat timeFormatter = new SimpleDateFormat("HHmmss");

			for (ThreadData thread : threads) {

				Document document = new Document();

				id.setValue(Identifier.generateId(thread.getBoardId(), thread.getThreadId()));
				boardId.setValue(thread.getBoardId() + "");
				title.setValue(thread.getTitle());
				name.setValue(thread.getDisplayName());
				login.setValue(thread.getLoginId());

				if (thread.isDeleted()) {
					deleted.setValue("true");
				} else {
					deleted.setValue("false");
				}

				date.setValue(formatter.format(thread.getTimestamp()));
				time.setValue(timeFormatter.format(thread.getTimestamp()));

				document.add(id);
				document.add(boardId);
				document.add(title);
				document.add(name);
				document.add(login);
				document.add(date);
				document.add(time);
				document.add(deleted);

				indexWriter.addDocument(document);
			}

			indexWriter.optimize();

		} catch (Exception e) {
			e.printStackTrace();

		} finally {
			indexWriter.close();
		}
	}


	/**
	 * (1) Get all the threads to be indexed and retrieve the most recent timestamp
	 * (2) Index the threads in Lucene
	 * (3) Mark all these threads as indexed in the database
	 */

	private void indexNewThreads() throws SQLException, IOException {

		List<ThreadData> threads = new LinkedList<ThreadData>();

		MutableDateTime mostRecentThreadTimestamp = this.getNewThreads(threads);
		this.indexThreads(threads);
		this.markIndexed(mostRecentThreadTimestamp, DocumentType.Threads);
	}


	private MutableDateTime getNewMessages(List<MessageData> messages) throws SQLException {

		Connection connection = null;

		try {

			MutableDateTime maxDateTime = new MutableDateTime(1, 1, 1, 1, 1, 1, 1);

			connection = Database.getConnection();
			PreparedStatement statement = connection.prepareStatement(BatchQueries.GET_MESSAGES_NEEDING_INDEXED);

			statement.execute();

			ResultSet resultSet = statement.getResultSet();

			while (resultSet.next()) {

				int boardId = resultSet.getInt("boardid");
				int threadId = resultSet.getInt("threadid");
				int messageId = resultSet.getInt("messageid");
				String text = resultSet.getString("body");
				String displayName = resultSet.getString("displayname");
				String loginId = resultSet.getString("userid");
				Timestamp timestamp = resultSet.getTimestamp("timestamp");
				boolean deleted = resultSet.getBoolean("deleted");

				DateTime time = new DateTime(timestamp);

				/* Ensure we keep a record of the most recent new thread */

				if (time.isAfter(maxDateTime)) {
					maxDateTime = time.toMutableDateTime();
				}

				MessageData message = new MessageData(boardId, threadId, messageId, text, loginId, displayName, timestamp, deleted);
				messages.add(message);
			}

			resultSet.close();
			statement.close();

			return maxDateTime;

		} finally {
			Database.close(connection);
		}

	}


	private void indexMessages(List<MessageData> messages) throws CorruptIndexException, IOException {

		if (messages.size() == 0) {
			return;
		}

		File indexDirectory = new File(Configuration.getInstance().getMessageIndexPath());
		IndexWriter indexWriter = new IndexWriter(indexDirectory, new StandardAnalyzer());

		try {
			Field id = new Field("id", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field boardId = new Field("boardId", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field text = new Field("text", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field name = new Field("displayName", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field login = new Field("loginId", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field date = new Field("date", "", Field.Store.YES, Field.Index.UN_TOKENIZED);
			Field time = new Field("time", "", Field.Store.YES, Field.Index.TOKENIZED);
			Field deleted = new Field("deleted", "", Field.Store.YES, Field.Index.UN_TOKENIZED);

			SimpleDateFormat formatter = new SimpleDateFormat("yyMMdd");
			SimpleDateFormat timeFormatter = new SimpleDateFormat("HHmmss");

			for (MessageData message : messages) {

				Document document = new Document();

				id.setValue(Identifier.generateId(message.getBoardId(), message.getThreadId(), message.getMessageId()));
				boardId.setValue(message.getBoardId() + "");
				text.setValue(message.getText());
				name.setValue(message.getDisplayName());
				login.setValue(message.getLoginId());

				if (message.isDeleted()) {
					deleted.setValue("true");
				} else {
					deleted.setValue("false");
				}

				date.setValue(formatter.format(message.getTimestamp()));
				time.setValue(timeFormatter.format(message.getTimestamp()));

				document.add(id);
				document.add(boardId);
				document.add(text);
				document.add(name);
				document.add(login);
				document.add(date);
				document.add(time);
				document.add(deleted);

				indexWriter.addDocument(document);
			}

		} finally {
			indexWriter.close();
		}

	}


	private void markIndexed(MutableDateTime mostRecentMessageTimestamp, DocumentType documentType) throws SQLException {

		Connection connection = null;

		try {

			String query = "";

			if (documentType == DocumentType.Threads) {
				query = BatchQueries.MARK_THREADS_INDEXED;
			} else if (documentType == DocumentType.Messages) {
				query = BatchQueries.MARK_MESSAGES_INDEXED;
			} else {
				throw new IllegalArgumentException("Pass in either DocumentType.Threads or DocumentType.Messages");
			}

			connection = Database.getConnection();

			PreparedStatement statement = connection.prepareStatement(query);
			statement.setTimestamp(1, new Timestamp(mostRecentMessageTimestamp.toDate().getTime()));

			statement.executeUpdate();

			statement.close();

		} finally {
			Database.close(connection);
		}

	}


	private void indexNewMessages() throws SQLException, CorruptIndexException, IOException {

		List<MessageData> messages = new LinkedList<MessageData>();

		MutableDateTime mostRecentMessageTimestamp = this.getNewMessages(messages);
		this.indexMessages(messages);
		this.markIndexed(mostRecentMessageTimestamp, DocumentType.Messages);

	}


	public void reindexAllDocuments() throws SQLException, IOException {
		
		/* Delete existing indices */
		
		FileUtils.deleteDirectory(new File(Configuration.getInstance().getMessageIndexPath()));
		FileUtils.deleteDirectory(new File(Configuration.getInstance().getThreadIndexPath()));
		
		this.setIndexAllThreads(true);
		this.indexNewThreads();
		
		Connection connection = null;
		
		try {

			connection = Database.getConnection();

			/* Get all the thread IDs into a list */

			PreparedStatement statement = connection.prepareStatement("SELECT boardid, threadid FROM threads");
			statement.execute();

			ResultSet resultSet = statement.getResultSet();

			List<int[]> threadIDs = new ArrayList<int[]>(100000);

			while (resultSet.next()) {
				int[] IDs = new int[2];
				IDs[0] = resultSet.getInt(1);
				IDs[1] = resultSet.getInt(2);
				threadIDs.add(IDs);
			}

			statement.close();

			/* Now index each thread */

			int messageCounter = 0;
			int total = 0;

			List<MessageData> messages = new ArrayList<MessageData>();

			for (int[] threadID : threadIDs) {

				int boardId = threadID[0];
				int threadId = threadID[1];

				statement = connection.prepareStatement("SELECT messageid, body, displayname, userid, timestamp, deleted FROM messages WHERE boardid=? AND threadid=?");
				statement.setInt(1, boardId);
				statement.setInt(2, threadId);

				statement.execute();

				resultSet = statement.getResultSet();

				while (resultSet.next()) {

					int messageId = resultSet.getInt("messageid");
					String text = resultSet.getString("body");
					String displayName = resultSet.getString("displayname");
					String loginId = resultSet.getString("userid");
					Timestamp timestamp = resultSet.getTimestamp("timestamp");
					boolean deleted = resultSet.getBoolean("deleted");

					MessageData message = new MessageData(boardId, threadId, messageId, text, loginId, displayName, timestamp, deleted);
					messages.add(message);

					messageCounter++;
				}

				resultSet.close();
				statement.close();

				/* Every 50,000 or so, index them */

				if (messageCounter >= 50000) {
					this.indexMessages(messages);
					total += messageCounter;
					messageCounter = 0;
					messages.clear();
					Logger.log(new Date(), "Indexed: " + total + " " + new Date(), User.getSystemUser(), LogMessageType.SYSTEM);
				}
			}

			/* Process remaining messages; there will be < 50,000 of them */

			this.indexMessages(messages);
			total += messageCounter;
			messageCounter = 0;
			messages.clear();
			Logger.log(new Date(), "Final processing, indexed: " + total + " " + new Date(), User.getSystemUser(), LogMessageType.SYSTEM);

			/* Optimise the message index */

			Logger.log(new Date(), "Optimising message index...", User.getSystemUser(), LogMessageType.SYSTEM);

			File indexDirectory = new File(Configuration.getInstance().getMessageIndexPath());
			IndexWriter indexWriter = new IndexWriter(indexDirectory, new StandardAnalyzer());
			indexWriter.optimize();
			indexWriter.close();

			/* Optimise the thread index */

			Logger.log(new Date(), "Optimising thread index...", User.getSystemUser(), LogMessageType.SYSTEM);

			indexDirectory = new File(Configuration.getInstance().getThreadIndexPath());
			indexWriter = new IndexWriter(indexDirectory, new StandardAnalyzer());
			indexWriter.optimize();
			indexWriter.close();

			/* Once fully indexed, mark all the messages as indexed in chunks of half a million */

			Logger.log(new Date(), "Marking messages as indexed...", User.getSystemUser(), LogMessageType.SYSTEM);

			int chunks = total / 500000;

			for (int i = 0; i < chunks; i++) {
				Logger.log(new Date(), "Updating chunk " + i + " of " + chunks, User.getSystemUser(), LogMessageType.SYSTEM);
				statement = connection.prepareStatement("UPDATE messages SET indexed=true WHERE indexed=false LIMIT 500000");
				statement.executeUpdate();
			}

			Logger.log(new Date(), "Indexing Completed...", User.getSystemUser(), LogMessageType.SYSTEM);

		} finally {
			Database.close(connection);
		}

	}
	
	
	/** 
	 *  Main method is for bulk indexing of lots of messages. Needs a lot of memory and relies on exclusive access to
	 *  the database. Do not have the site running when this is running. Intended as an initial one off job, which
	 *  allows you to index a large database of messages in one go, since the batch methods in this class as not
	 *  suitable for large volumes of data (i.e. those which could blow the transaction log). */

	public static void main(String[] args) throws Exception {

		IndexNewDocuments indexNewDocuments = new IndexNewDocuments();

		/* Make sure you configure this correctly */

		Database.url = "jdbc:mysql://localhost:3306/ilxtest";
		Configuration.getInstance().setConnectionStrategy(ConnectionStrategy.StaticConnection);
		Configuration.load();
		Configuration.getInstance().setConnectionStrategy(ConnectionStrategy.StaticConnection);

		System.out.println("The database URL is: " + Database.url + ". Are you happy with this? Stop the program if not.");
		System.in.read();

		System.out.println("Do you want to index everything? Press a key to continue...");
		System.in.read();

		indexNewDocuments.reindexAllDocuments();

	}

}
